package com.bianmaba.jobs.collection.monitor;

import com.alibaba.fastjson.JSON;
import com.bianmaba.dc.bean.CollectionData;
import com.bianmaba.dc.bean.InterfaceType;
import com.bianmaba.dc.bean.RedisKeys;
import com.bianmaba.jobs.collection.processor.DtsDataProcessor;
import com.bianmaba.services.datacollection.quartiz.QuartzTaskManager;
import lombok.extern.log4j.Log4j2;
import org.apache.commons.lang3.StringUtils;
import org.quartz.*;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Bean;
import org.springframework.data.redis.core.ListOperations;
import org.springframework.scheduling.quartz.SimpleTriggerFactoryBean;
import org.springframework.stereotype.Component;

/**
 * @program: samples
 * @description:
 * @author: Chenjiabin
 * @create: 2019/10/8 15:54
 **/
@Component
@Log4j2
@DisallowConcurrentExecution
public class PendingDataMonitor implements Job {
    @Autowired
    protected ListOperations<String, Object> redisListOperations;
    @Autowired
    protected QuartzTaskManager quartzTaskManager;

    @Override
    public void execute(JobExecutionContext jobExecutionContext) throws JobExecutionException {
        log.info("采集数据接收器开始运行...");
        try {
            int count = 1;
            while (true) {
                Object obj = redisListOperations.rightPop(RedisKeys.PENDING_DATA);
                if (obj != null && StringUtils.isNotEmpty(obj.toString())) {
                    CollectionData datas = JSON.parseObject(obj.toString(), CollectionData.class);
                    handle(datas);
                    Long size = redisListOperations.size("collection_datas");
                    log.info("采集数据接收器待处理数据" + size + "条");
                    count = 1;
                } else {
                    Thread.sleep(count * 1000);
                    if (count < 5) {
                        count++;
                    }
                }
            }
        } catch (InterruptedException e) {
            log.info("采集数据接收器终止运行");
        }
    }

    private void handle(CollectionData datas) {
        if (datas == null) {
            log.error("待处理数据为空。");
        }

        String interfaceType = datas.getInterfaceType();
        if (InterfaceType.DTS.equalsIgnoreCase(interfaceType)) {
            String projectId = datas.getProjectId();
            String taskId = datas.getTaskId();
            String taskInstanceId = datas.getTaskInstanceId();
            if (StringUtils.isEmpty(projectId)) {
                log.error("采集数据中缺少项目编号，无法处理采集数据。");
                return;
            }
            if (StringUtils.isEmpty(taskId)) {
                log.error("采集数据中缺少任务编号，无法处理采集数据。");
                return;
            }
            if (StringUtils.isEmpty(taskInstanceId)) {
                log.error("采集数据中缺少任务实例编号，无法处理采集数据。");
                return;
            }
            JobDataMap jobData = new JobDataMap();
            jobData.put("datas", datas);
            quartzTaskManager.executeOnceJob(DtsDataProcessor.class, projectId, taskInstanceId, jobData);
        } else {
            log.error("接口类型无法识别，无法处理采集数据。");
        }
    }

    @Bean
    public SimpleTriggerFactoryBean startRevicer() {
        JobDetail jobDetail = JobBuilder.newJob(PendingDataMonitor.class)
                .withIdentity("COLLECTION_DATA_RECEIVER").build();
        SimpleTriggerFactoryBean trigger = new SimpleTriggerFactoryBean();
        trigger.setJobDetail(jobDetail);
        trigger.setRepeatInterval(1000);
        trigger.setRepeatCount(Integer.MAX_VALUE);
        return trigger;
    }
}
